package com.lxw.oa.util;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.annotation.JSONField;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.sql.*;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by Administrator on 2019/4/22.
 */
public class DataxUtil {

    static {
        try {
            Class.forName("com.mysql.jdbc.Driver");
        } catch (ClassNotFoundException e) {
            System.out.println("加载驱动异常");
            e.printStackTrace();
        }
    }


    public static  Connection conn = null;
    public static  Statement stmt = null;
    public static  ResultSet rs = null;
    public static  String url="";
    //所有的建表语句都放在一起
    public static  String HiveSqlAll="";

    //所有的删表删库语句放在一起
    public static  String HiveDelAll="";

    //所有的建表语句都放在一起
    public static  String JsonShell="cd /usr/local/datax/;\n";


    //根据数据库名循环对每张表进行转换
    public void DoneAll(String ip,String port,String database,String userName,String passWord){
        HiveSqlAll+="CREATE DATABASE IF NOT EXISTS "+database+";\n";
        HiveDelAll+="use "+database+";\n";
        HiveSqlAll+="use "+database+";\n";
        SimpleDateFormat formatter = new SimpleDateFormat("yyyyMMddHHmmss");
        String currentTime = formatter.format(System.currentTimeMillis());

        //这个为生成的·json与sql文件的存储路径。
        String jsonFilepath = "D:\\DataXJsons\\"+database+"-"+currentTime;
        String sqlFilepath = "D:\\DataXHiveSqls\\"+database+"-"+currentTime;

        url="jdbc:mysql://"+ip+":"+port+"/"+database+"?user="+userName+"&password="+passWord;
        try {
            conn = DriverManager.getConnection(url); // 获取数据库连接
            stmt = conn.createStatement(); // 创建执行环境
            String sqlStr="select table_name from information_schema.tables where table_schema='"+database+"' " +
                    "and table_type='base table' ";
            rs = stmt.executeQuery(sqlStr); // 执行查询语句，返回结果数据集
            //所有的数据库表名的集合
            List<String> tableNameList=new ArrayList<String>();
            while (rs.next()) { // 循环读取结果数据集中的所有记录
                String tableName=rs.getString(1);
                tableNameList.add(tableName);
            }

            for (String tableName: tableNameList) {
                DoneOne(ip,port,database,userName,passWord,tableName,jsonFilepath,sqlFilepath);
            }

            HiveDelAll+="DROP DATABASE IF EXISTS "+database+";\n";

            //输出所有的Hive删表删库语句
            fileOutPutSql(sqlFilepath,HiveDelAll,"del","del");

            //输出所有的Hive建表语句
            fileOutPutSql(sqlFilepath,HiveSqlAll,"all","all");

            //输出所有的执行Shell命令
            fileOutPutJson(jsonFilepath,JsonShell,"all","all");

        }catch(SQLException e) {
        System.out.println("数据库异常");
        e.printStackTrace();
        } finally {
            try {
                if (rs != null)
                    rs.close(); // 关闭结果数据集
                if (stmt != null)
                    stmt.close(); // 关闭执行环境
                if (conn != null)
                    conn.close(); // 关闭数据库连接
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
    };

    //根据数据库名，表名对单表进行转换
    public void DoneOne(String ip,String port,String database,String userName,String passWord,String tableName,String jsonFilepath,String sqlFilepath){
        String url="jdbc:mysql://"+ip+":"+port+"/"+database;
        List<Map<String,String>> list=getColList(ip,port,database,userName,passWord,tableName);
        String columnStr="";
        for(int i=0;i<list.size();i++){
                if(i<list.size()-1){
                    columnStr+="`"+list.get(i).get("name")+"`,";
                }else{
                    columnStr+="`"+list.get(i).get("name")+"`";
                }

        }
        String sql="select "+columnStr+" from "+tableName;
        String path="/user/hive/warehouse/"+database+".db/"+tableName;
        String fileName=tableName;
        String jsonStr=toJson(userName,passWord,url,sql,path,fileName,list);
        String createTableStr=getCreateTableStr(tableName,list);
        HiveDelAll+="drop table if exists "+tableName+";\n";
/*        System.out.println(jsonStr);
        System.out.println(createTableStr);*/
        fileOutPutJson(jsonFilepath,jsonStr,database,tableName);
        fileOutPutSql(sqlFilepath,createTableStr,database,tableName);
    }

    //得到对应表的可以传输的数据库的 字段列表,并调用tranType方法转换字段类型
    public static List<Map<String,String>> getColList(String ip,String port,String database,String userName,String passWord,String tableName){
        List<Map<String,String>> list=new ArrayList<Map<String, String>>();
        String sqlStr="select column_name,data_type,column_comment from information_schema.columns " +
                "where table_schema ='"+database+"'  and table_name = '"+tableName+"'";
        try{
            rs = stmt.executeQuery(sqlStr); // 执行查询语句，返回结果数据集
//            rs.last(); // 将光标移到结果数据集的最后一行，用来下面查询共有多少行记录
//            System.out.println("共有" + rs.getRow() + "个字段：");
//            rs.beforeFirst(); // 将光标移到结果数据集的开头
            while (rs.next()) { // 循环读取结果数据集中的所有记录
                String name=rs.getString(1);
                String type=rs.getString(2);
                String comment=rs.getString(3);
                if(comment==null){
                    comment="";
                }
                String hiveType=tranType(type);

                //如果该列的字段类型无法被转换成功，则不同步该列的数据。
                if("".equals(hiveType)){

                }else{
                    Map<String,String> map=new HashMap<>();
                    map.put("name",name);
                    map.put("type",hiveType);
                    map.put("comment",comment);
                    list.add(map);
                }


            }
        }catch(SQLException e) {
            System.out.println("数据库异常");
            e.printStackTrace();
        } finally {
            try {
                if (rs != null)
                    rs.close(); // 关闭结果数据集
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        return list;
    }

    //转换字段类型
    public static String tranType(String type){
        type=type.toLowerCase();

        switch(type){
            case "int":
                type="int";
                break;
            case "tinyint":
                type="tinyint";
                break;
            case "smallint":
                type="smallint";
                break;
            case "mediumint":
                type="int";
                break;
            case "bigint":
                type="bigint";
                break;
            case "float":
                type="float";
                break;
            case "double":
                type="double";
                break;
            case "decimal":
                type="double";
                break;
            case "varchar":
                type="string";
                break;
            case "char":
                type="string";
                break;
            case "tinytext":
                type="string";
                break;
            case "mediumtext":
                type="string";
                break;
            case "longtext":
                type="string";
                break;
            case "year":
                type="string";
                break;
            case "date":
                type="timestamp";
                break;
            case "datetime":
                type="timestamp";
                break;
            case "timestamp":
                type="timestamp";
                break;
            default:
                type="";
                break;
        }
        return type;
    }

    //得到Hive建表语句
    public static String getCreateTableStr(String tableName,List<Map<String,String>> list){
        String sqlStr="create table "+tableName+"(\n";
        String columnStr="";

        for(int i=0;i<list.size();i++){
            if(i<list.size()-1){
                columnStr+=list.get(i).get("name")+" "+list.get(i).get("type")+",\n";
            }else{
                columnStr+=list.get(i).get("name")+" "+list.get(i).get("type");
            }

        }
        sqlStr+=columnStr+"\n)row format delimited fields terminated by \"\\t\" STORED AS TEXTFILE;";
        HiveSqlAll+=sqlStr+"\n";
        return sqlStr;
    }

    //将数据转成json
    public String toJson(String userName,String passWord,String url,String sql,String path,String fileName,List<Map<String,String>> list){

        JsonPojo jsonPojo=new JsonPojo();
        JobPojo jobPojo=new JobPojo();
        jsonPojo.setJobPojo(jobPojo);

        //================>Setting部分
        SettingPojo settingPojo=new SettingPojo();
        jobPojo.setSettingPojo(settingPojo);

        SpeedPojo speedPojo=new SpeedPojo();
        settingPojo.setSpeedPojo(speedPojo);
        speedPojo.setChannelPojo("5");

        //================>Content部分
        List<ContentPojo>contentPojoList=new ArrayList<ContentPojo>();
        jobPojo.setContentPojoList(contentPojoList);

        ContentPojo contentPojo=new ContentPojo();
        contentPojoList.add(contentPojo);

        //================>Content->Reader部分
        ReaderPojo readerPojo=new ReaderPojo();
        contentPojo.setReaderPojo(readerPojo);

        ParameterPojo parameterPojo=new ParameterPojo();
        readerPojo.setParameterPojo(parameterPojo);
        parameterPojo.setUsernamePojo(userName);
        parameterPojo.setPasswordPojo(passWord);

        List<ConnectionPojo> ConnectionList=new ArrayList<ConnectionPojo>();
        parameterPojo.setConnectionPojoList(ConnectionList);

        ConnectionPojo connectionPojo=new ConnectionPojo();
        ConnectionList.add(connectionPojo);

        List<String> querySql=new ArrayList<String>();
        List<String> jdbcUrl=new ArrayList<String>();
        querySql.add(sql);
        jdbcUrl.add(url);

        connectionPojo.setQuerySql(querySql);
        connectionPojo.setJdbcUrl(jdbcUrl);


        //================>Content->Writer部分
        WriterPojo writerPojo=new WriterPojo();
        contentPojo.setWriterPojo(writerPojo);

        ParameterWPojo parameterWPojo=new ParameterWPojo();
        writerPojo.setParameter(parameterWPojo);
        parameterWPojo.setFileName(fileName);
        parameterWPojo.setPath(path);


        List<ColumnPojo>columnPojoList=new ArrayList<ColumnPojo>();
        parameterWPojo.setColumnPojoList(columnPojoList);

        for (Map<String,String> map: list) {
            ColumnPojo columnPojo=new ColumnPojo();
            columnPojo.setName(map.get("name"));
            columnPojo.setType(map.get("type"));
            columnPojoList.add(columnPojo);
        }

        return JSON.toJSONString(jsonPojo);
    }

    //导出json文件
    public static void  fileOutPutJson(String filepath,String txt,String database,String tableName){
        File file = new File(filepath);
        if(!file.exists()){
            file.mkdirs();
        }
        File f = new File(file,tableName+".json");

        //添加传输的shell脚本文件
        JsonShell+="bin/datax.py job/"+database+"/"+f.getName()+";\n                 \n";

        FileWriter fw = null;
        try {
            fw = new FileWriter(f, true);
        } catch (IOException e) {
            e.printStackTrace();
        }
        BufferedWriter bw = new BufferedWriter(fw);
        try {
            bw.write(txt);
        } catch (IOException e) {
            e.printStackTrace();
        }
        try {
            bw.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
        try {
            bw.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        try {
            fw.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    //导出sql文件
    public static void  fileOutPutSql(String filepath,String txt,String database,String tableName){
        File file = new File(filepath);
        if(!file.exists()){
            file.mkdirs();
        }
        File f = new File(file,tableName+".sql");
//        try {
//            f = File.createTempFile(fileName, ".sql",file);
//        } catch (IOException e) {
//            e.printStackTrace();
//        }

        FileWriter fw = null;
        try {
            fw = new FileWriter(f, true);
        } catch (IOException e) {
            e.printStackTrace();
        }
        BufferedWriter bw = new BufferedWriter(fw);
        try {
            bw.write(txt);
        } catch (IOException e) {
            e.printStackTrace();
        }
        try {
            bw.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
        try {
            bw.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        try {
            fw.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //总的Json
    class JsonPojo{

        @JSONField(name="job")
        private JobPojo jobPojo;

        public JobPojo getJobPojo() {
            return jobPojo;
        }

        public void setJobPojo(JobPojo jobPojo) {
            this.jobPojo = jobPojo;
        }
    }

    class JobPojo{
        @JSONField(name="setting")
        private SettingPojo settingPojo;

        @JSONField(name="content")
        private List<ContentPojo> contentPojoList;

        public SettingPojo getSettingPojo() {
            return settingPojo;
        }

        public void setSettingPojo(SettingPojo settingPojo) {
            this.settingPojo = settingPojo;
        }

        public List<ContentPojo> getContentPojoList() {
            return contentPojoList;
        }

        public void setContentPojoList(List<ContentPojo> contentPojoList) {
            this.contentPojoList = contentPojoList;
        }
    }

    //================>Setting部分
    class SettingPojo{
        @JSONField(name="speed")
        private SpeedPojo speedPojo;

        public SpeedPojo getSpeedPojo() {
            return speedPojo;
        }

        public void setSpeedPojo(SpeedPojo speedPojo) {
            this.speedPojo = speedPojo;
        }
    }

    class SpeedPojo{
        @JSONField(name="channel")
        private String channelPojo;

        public String getChannelPojo() {
            return channelPojo;
        }

        public void setChannelPojo(String channelPojo) {
            this.channelPojo = channelPojo;
        }
    }

    //================>Content部分
    class ContentPojo{
        @JSONField(name="reader")
        private ReaderPojo readerPojo;
        @JSONField(name="writer")
        private WriterPojo writerPojo;

        public ReaderPojo getReaderPojo() {
            return readerPojo;
        }

        public void setReaderPojo(ReaderPojo readerPojo) {
            this.readerPojo = readerPojo;
        }

        public WriterPojo getWriterPojo() {
            return writerPojo;
        }

        public void setWriterPojo(WriterPojo writerPojo) {
            this.writerPojo = writerPojo;
        }
    }

    //================>Content->Reader部分
    class ReaderPojo{
        @JSONField(name="name")
        private String namePojo ="mysqlreader";

        @JSONField(name="parameter")
        private ParameterPojo parameterPojo;

        public ParameterPojo getParameterPojo() {
            return parameterPojo;
        }

        public void setParameterPojo(ParameterPojo parameterPojo) {
            this.parameterPojo = parameterPojo;
        }

        public String getNamePojo() {
            return namePojo;
        }

        public void setNamePojo(String namePojo) {
            this.namePojo = namePojo;
        }
    }

    class ParameterPojo{
        @JSONField(name="username")
        private String usernamePojo;
        @JSONField(name="password")
        private String passwordPojo;
        @JSONField(name="connection")
        private List<ConnectionPojo> connectionPojoList;

        public String getUsernamePojo() {
            return usernamePojo;
        }

        public void setUsernamePojo(String usernamePojo) {
            this.usernamePojo = usernamePojo;
        }

        public String getPasswordPojo() {
            return passwordPojo;
        }

        public void setPasswordPojo(String passwordPojo) {
            this.passwordPojo = passwordPojo;
        }

        public List<ConnectionPojo> getConnectionPojoList() {
            return connectionPojoList;
        }

        public void setConnectionPojoList(List<ConnectionPojo> connectionPojoList) {
            this.connectionPojoList = connectionPojoList;
        }
    }

    class ConnectionPojo{

        @JSONField(name="querySql")
        private List<String> querySql;

        @JSONField(name="jdbcUrl")
        private List<String> jdbcUrl;

        public List<String> getQuerySql() {
            return querySql;
        }

        public void setQuerySql(List<String> querySql) {
            this.querySql = querySql;
        }

        public List<String> getJdbcUrl() {
            return jdbcUrl;
        }

        public void setJdbcUrl(List<String> jdbcUrl) {
            this.jdbcUrl = jdbcUrl;
        }
    }

    //================>Content->Writer部分
    class WriterPojo{
        @JSONField(name="name")
        private String name="hdfswriter";

        @JSONField(name="parameter")
        private ParameterWPojo parameter;

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public ParameterWPojo getParameter() {
            return parameter;
        }

        public void setParameter(ParameterWPojo parameter) {
            this.parameter = parameter;
        }
    }

    class ParameterWPojo{
        @JSONField(name="defaultFS")
        private String defaultFS="hdfs://192.168.5.69:8020";

        @JSONField(name="fileType")
        private String fileType="text";

        @JSONField(name="path")
        private String path;

        @JSONField(name="fileName")
        private String fileName;

        @JSONField(name="column")
        private List<ColumnPojo> columnPojoList;

        @JSONField(name="writeMode")
        private String writeMode="append";

        @JSONField(name="fieldDelimiter")
        private String fieldDelimiter="\t";

        @JSONField(name="compress")
        private String compress="GZIP";

        public String getDefaultFS() {
            return defaultFS;
        }

        public void setDefaultFS(String defaultFS) {
            this.defaultFS = defaultFS;
        }

        public String getFileType() {
            return fileType;
        }

        public void setFileType(String fileType) {
            this.fileType = fileType;
        }

        public String getPath() {
            return path;
        }

        public void setPath(String path) {
            this.path = path;
        }

        public String getFileName() {
            return fileName;
        }

        public void setFileName(String fileName) {
            this.fileName = fileName;
        }

        public List<ColumnPojo> getColumnPojoList() {
            return columnPojoList;
        }

        public void setColumnPojoList(List<ColumnPojo> columnPojoList) {
            this.columnPojoList = columnPojoList;
        }

        public String getWriteMode() {
            return writeMode;
        }

        public void setWriteMode(String writeMode) {
            this.writeMode = writeMode;
        }

        public String getFieldDelimiter() {
            return fieldDelimiter;
        }

        public void setFieldDelimiter(String fieldDelimiter) {
            this.fieldDelimiter = fieldDelimiter;
        }

        public String getCompress() {
            return compress;
        }

        public void setCompress(String compress) {
            this.compress = compress;
        }
    }

    class ColumnPojo{
        @JSONField(name="name")
        private String name;

        @JSONField(name="type")
        private String type;

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public String getType() {
            return type;
        }

        public void setType(String type) {
            this.type = type;
        }
    }

    public static void main(String[] args){
/*        List<Map<String,String>> list=DataxUtil.getColList("47.107.121.240","3306","oamanager","root","147512","t_sys_user");
        for (Map<String,String> map :list) {
            System.out.println("=================>");
            System.out.println(map.get("name"));
            System.out.println(map.get("type"));
            System.out.println(map.get("comment"));
        }*/
        new DataxUtil().DoneAll("47.107.121.240","3306","impala_test","root","147512");

    }

}
